/*
 * Copyright (C) 2010 TopCoder Inc., All Rights Reserved.
 */
package com.hercules.xre.protocol.testing.netty;



import org.jboss.netty.bootstrap.ServerBootstrap;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelException;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import com.hercules.xre.protocol.testing.EventHandler;
import com.hercules.xre.protocol.testing.Helper;
import com.hercules.xre.protocol.testing.ProtocolTestingConnectionManager;
import com.hercules.xre.protocol.testing.ProtocolTestingConnectionManagerException;
import com.topcoder.util.log.Log;
import com.topcoder.util.log.LogManager;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


import javax.net.ssl.SSLEngine;
/**
 * <p>
 * Base class for netty protocol testing connection managers.
 * </p>
 * <p>
 * As most of the functionality that ProtocolTestingConnectionManager provides is done by ChannelHandlers,
 * only initialization of channel handlers pipeline should be delegated to child classes (getPipelineFactory()).
 * </p>
 * <p>
 * In order to use communication encryption appropriate SSLEngineFactory has
 * to be created and passed to setSSLEngineFactory() method.
 * </p>
 * <p>
 * This class provides also two maps that can be used to map channel to a ssessionGUID (in both directions),
 * two maps have been used in order to perform all operations effectively. Mentioned maps will be used by channel
 * handlers and thus they can be accessed from multiple threads. In order to provide appropriate synchronization all
 * access to mentioned maps should be done in "synchronized" block. Synchronization should be done on
 * "associationMapsLock" object).
 * </p>
 * <p>
 * Sample Usage:
 * int port = 1234;
 * // create new instance of connection manager
 * // connection manager won't be using encryption
 * BaseProtocolTestingConnectionManager connectionManager = new BaseProtocolTestingConnectionManager(port, null, null);
 *
 * connectionManager.startAcceptingConnections(eventHandler);
 *
 * // wait for clients to connect
 *
 * // when there is a session opened a command can be sent
 * String sessionGUID = &quot;1&quot;
 * <br>
 * String command = "{command: \"CONNECT\",version: \"1.2.3\",sessionGUID: \"1\",
 * commandIndex: 0,timestamp: 123456789}";
 * </br>
 * connectionManager.addChannelSessionGUIDAssociation(channel, sessionGUID);
 * connectionManager.sendCommand(sessionGUID, command);
 *
 * // whenever new message is received a eventHandler will be called
 *
 * // when connection manager is not needed anymore, dispatch() should be called
 * connectionManager.dispose();
 * </p>
 * <p>
 * <strong>Thread-safety:</strong> This class isn't thread safe but it exposes 4 methods that are thread safe and
 * can be used by multiple threads, mentioned methods are: addChannelSessionGUIDAssociation, getSessionGUIDByChannel,
 * getChannelBySessionGUID, removeChannelSessionGUID.
 * </p>
 * <p>
 * Mentioned methods are used by channel handlers to get/set/remove association between a channel and a session GUID.
 * All of them use proper synchronization when association maps are accessed. These maps are also accessed in
 * dispose() method, but synchronization is also used in there.
 * </p>
 * @author TCSDESIGNER, TCSDEVELOPER
 * @version 1.0
*/
public abstract class BaseProtocolTestingConnectionManager implements ProtocolTestingConnectionManager {
    /**
     * <p>
     * Logger that is used to log activities and errors.
     * </p>
     * <p>
     * It is initialized to LogManager.getLog(BaseProtocolTestingConnectionManager.class.getName()).
     * Can not be null, never changes after initialization.
     * Subclass could access this logger to perform additional logging.
     * </p>
     */
    protected static final Log LOGGER = LogManager.getLog(BaseProtocolTestingConnectionManager.class.getName());

    /**
     * <p>
     * Pattern that is used to match "commandIndex" in the command string.
     * </p>
     * <p>
     * It is initialized to Pattern.compile("commandIndex[\\s]*:[\\s]*[\\d]+").
     * Can not be null, never changes after initialization.
     * </p>
     */
    private static final Pattern COMMAND_INDEX_PATTERN = Pattern.compile("\"commandIndex\"[\\s]*:[\\s]*[\\d]+");

    /**
     * <p>
     * Pattern that is used to match "timestamp" in the command string.
     * </p>
     * <p>
     * It is initialized to Pattern.compile("timestamp[\\s]*:[\\s]*[\\d]+").
     * Can not be null, never changes after initialization.
     * </p>
     */
    private static final Pattern TIMESTAMP_PATTERN = Pattern.compile("\"timestamp[\\s]*\":[\\s]*[\\d]+");

    /**
     * <p>
     * Port number on which application should listen for incoming connections.
     * </p>
     * <p>
     * Initialized in constructor. A valid port value is between 1 and 65535.
     * </p>
     */
    private final int  port;

    /**
     * <p>
     * Address on which application should listen for incoming connections.
     * </p>
     * <p>
     * Initialized in constructor. Can be null. If set to null then application will be
     * listening on all interfaces.
     * </p>
     */
    private final String  bindAddress;

    /**
     * <p>
     * SSLEngine factory instance that will be used to create encryption engines.
     * It is set in setSSLEngineFactory() method. Can be null.
     * </p>
     * <p>
     * If it is set to null then encryption is considered to be disabled.
     * </p>
     */
    private SSLEngineFactory sslEngineFactory;

    /**
     * <p>
     * Object on which synchronization is performed. All access to channel<->sessionGUID
     * maps is wrapped with synchronization blocks over this object.
     * </p>
     * <p>
     * Can't be null. Never changes after initialization.
     * </p>
     */
    private final Object  associationMapsLock =  new Object();

    /**
     * <p>
     * Map used to represent association between channel and it's session GUID.
     * </p>
     * <p>
     * It can be used in multi-threaded environment and thus all access to it should be
     * wrapped within synchronized block (synchronization should be done on associationMapsLock object).
     * </p>
     * <p>
     * Can't be null. Values can be added or removed after initialization.
     * Values should represent not null Channel instances. Keys should be not null,
     * not empty strings.
     * </p>
     */
    private final Map<Channel, String> channelSessionGUIDMap = new HashMap<Channel, String>();

    /**
     * <p>
     * Map used to represent association between session GUID and appropriate channel.
     * </p>
     * <p>
     * It can be used in multi-threaded environment and thus all access to it should be
     * wrapped within synchronized block (synchronization should be done on associationMapsLock object).
     * </p>
     * <p>
     * Can't be null. Values can be added or removed after initialization.
     * Keys should be not null, not empty strings, and values should represent not null Channel instances.
     * </p>
     */
    private final Map<String, Channel> sessionGUIDChannelMap =  new HashMap<String, Channel>();

    /**
     * <p>
     * Map used to represent a mapping from session GUID to the next command index for the sessionGUID.
     * </p>
     * <p>
     * It can be used in multi-threaded environment and thus all access to it should be wrapped
     * within synchronized block (synchronization should be done on associationMapsLock object).
     * </p>
     * <p>
     * Can't be null. Values can be added or removed after initialization.
     * Keys should be not null, not empty strings, and values should be non-null, non-negative Integer's.
     * </p>
     */
    private final Map<String, Integer> sessionGUIDNextCommandIndexMap = new HashMap<String, Integer>();

    /**
     * <p>
     * Channel that is used by application to listen for incoming connections.
     * </p>
     * <p>
     * It is initialized in startAcceptingConnections().
     * Can be null. Shouldn't be null after a call to startAcceptingConnections()
     * </p>
     */
    private Channel serverChannel;

    /**
     * <p>
     * ChannelFactor that is used to create Channel instances for connections to application.
     * </p>
     * <p>
     * It is initialized in startAcceptingConnections().
     * Can be null, shouldn't be null after a call to startAcceptingConnections()
     * </p>
     */
    private ChannelFactory channelFactory;

    /**
     * <p>
     * Creates new instance with passed values.
     * </p>
     *
     * @param sslEngineFactory ssl engine factory, can be null (if null no encryption will be used).
     * @param bindAddress address to which application should be bound, if null is passed then
     *        application will bind on all available interfaces
     * @param port port number on which connection manager should be listening.
     *        A valid port value is between 1 and 65535.
     * @throws IllegalArgumentException when any of passed parameters is invalid
     */
    public BaseProtocolTestingConnectionManager(int port, String bindAddress, SSLEngineFactory sslEngineFactory) {
        final String signature = "BaseProtocolTestingConnectionManager#ctor";
        try {
            Helper.checkPortAndLog(signature, new String[] {"port", "bindAddress", "sslEngineFactory"},
                new Object[] {port, bindAddress, sslEngineFactory}, LOGGER);
            Helper.checkEmpty(bindAddress, "bindAddress", signature, LOGGER);
            this.port = port;
            this.bindAddress = bindAddress;
            this.sslEngineFactory = sslEngineFactory;
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Setter for sslEngineFactory.
     * </p>
     * @param sslEngineFactory ssl engine factory
     */
    public void setSSLEngineFactory(SSLEngineFactory sslEngineFactory) {
        this.sslEngineFactory = sslEngineFactory;
    }

    /**
     * <p>
     * Method responsible for returning a SSLEngine that should be used for encryption.
     * If encryption shouldn't be used, null is returned.
     * </p>
     * @return if encryption should be used, returns null if there is no SSLEngineFactory set.
     */
    public SSLEngine getSSLEngine() {
        final String signature = "BaseProtocolTestingConnectionManager#getSSLEngine";
        Helper.logEnter(signature, null, null, LOGGER);
        return Helper.logExit(signature, sslEngineFactory != null ? sslEngineFactory
            .createEngine() : null, LOGGER);
    }

    /**
     * <p>
     * Method responsible for starting accepting incoming connections.
     * </p>
     *
     * <p>
     * This method calls subclass getPipelineFactory() to get pipeline factory it also binds
     * application to specified address/port and starts listening.
     * </p>
     *
     * @param eventHandler the event handler that should be used to inform
     *        application about communication events, can't be null
     * @throws IllegalArgumentException - when passed parameter is invalid
     * @throws NettyProtocolTestingConnectionManagerException when application
     *         can't bind to chosen port or any other exception is thrown when initializing
     */
    public void startAcceptingConnections(EventHandler eventHandler)
        throws NettyProtocolTestingConnectionManagerException {
        final String signature = "BaseProtocolTestingConnectionManager#startAcceptingConnections";
        Helper.logEnter(signature, new String[] {"eventHandler"}, new Object[] {eventHandler}, LOGGER);
        try {
            Helper.checkNull(eventHandler, "eventHandler", signature, LOGGER);
            // check if connection manager isn't already listening
            if (isStarted()) {
                throw Helper.logError(new ProtocolTestingConnectionManagerAlreadyRunningException(
                    "manager was already started"), signature, LOGGER);
            }

            channelFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
                Executors.newCachedThreadPool());
            ServerBootstrap bootstrap = new ServerBootstrap(channelFactory);
            bootstrap.setPipelineFactory(getPipelineFactory(eventHandler));

            SocketAddress bindSocketAddress = null;
            if (bindAddress != null && !bindAddress.trim().equals("")) {
                // bind only on a specified address
                try {
                    bindSocketAddress = new InetSocketAddress(InetAddress.getByName(bindAddress), port);
                } catch (UnknownHostException e) {
                    throw Helper.logError(new NettyProtocolTestingConnectionManagerException(
                        "Unkown hotst to initialize the address", e), signature, LOGGER);
                } catch (SecurityException e) {
                    throw Helper.logError(new NettyProtocolTestingConnectionManagerException(
                        "security manager doesn't allow getByName", e), signature, LOGGER);
                }
            } else {
                bindSocketAddress = new InetSocketAddress(port);
            }

            // Bind and start to accept incoming connections.
            try {
                serverChannel = bootstrap.bind(bindSocketAddress);
            } catch (ChannelException e) {
                throw Helper.logError(new NettyProtocolTestingConnectionManagerException(
                    "Can not bind the address", e), signature, LOGGER);
            }
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Method that should be implemented by all concrete subclasses.
     * Returns ChannelPipelineFactory instance that should be used for initialization of channels pipeline.
     * </p>
     *
     * @param eventHandler event handler that should be used to inform application about communication events,
     *        can't be null
     * @return ChannelPipelineFactory instance.
     * @throws IllegalArgumentException if eventHandler is null.
     */
    protected abstract ChannelPipelineFactory getPipelineFactory(EventHandler eventHandler);

    /**
     * <p>
     * Method responsible for sending passed command to appropriate client.
     * It first gets channel to which command should be sent and after that a write event
     * is passed to retrieved channels pipeline.
     * </p>
     *
     * @param sessionGUID GUID of a session, can't be null.
     * @param command command that should be sent, can't be null.
     * @throws llegalArgumentException when any of passed parameters is invalid
     * @throws ProtocolTestingConnectionManagerNotRunningException when connection manager hasn't been started
     * @throws NoSuchSessionException if can not find the session
     */
    public void sendCommand(String sessionGUID, String command)
        throws ProtocolTestingConnectionManagerNotRunningException, NoSuchSessionException {
        final String signature = "BaseProtocolTestingConnectionManager#sendCommand";
        Helper.logEnter(signature, new String[] {"sessionGUID", "command"},
            new Object[] {sessionGUID, command}, LOGGER);
        try {
            Helper.checkString(sessionGUID, "sessionGUID", signature, LOGGER);
            Helper.checkString(command, "command", signature, LOGGER);
            if (!isStarted()) {
                // Server has not been started
                throw Helper.logError(new ProtocolTestingConnectionManagerNotRunningException(
                    "Server has not been started"), signature, LOGGER);
            }
            Channel channel;
            try {
                channel = getChannelBySessionGUID(sessionGUID);
                Helper.checkNull(channel.getPipeline(), "pipeline", signature, LOGGER);
                ChannelFuture future = Channels.write(channel, command);
                future.addListener(new ExceptionHandlingChannelFutureListener());
            } catch (NoSuchSessionException e) {
                throw Helper.logError(e, signature, LOGGER);
            }
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Closes all connections and disposes other referenced resources.
     * </p>
     * @throws ProtocolTestingConnectionManagerException - if ConnectionManager
     *         hasn't been initialized (serverChannel and channelFactory are set to null)
     * @throws ProtocolTestingConnectionManagerNotRunningException - thrown when connection
     *         manager hasn't been started
     */
    public void dispose() throws ProtocolTestingConnectionManagerNotRunningException {
        final String signature = "BaseProtocolTestingConnectionManager#dispose";
        Helper.logEnter(signature, null, null, LOGGER);
        try {
            if (!isStarted()) {
                // Server has not been started
                throw Helper.logError(new ProtocolTestingConnectionManagerNotRunningException(
                    "Server has not been started"), signature, LOGGER);
            }
            ChannelGroup openedChannels = new DefaultChannelGroup();
            openedChannels.add(serverChannel);

            synchronized (associationMapsLock) {
                // close all child channels
                for (Channel channel : channelSessionGUIDMap.keySet()) {
                    // Note : ChannelGroup#add will throw NPE if id or close future of the channel is null
                    if (channel.getId() == null) {
                        Helper.logError(new ProtocolTestingConnectionManagerException(
                            "the id of channel can't be null"), signature, LOGGER);
                    }
                    if (channel.getCloseFuture() == null) {
                        Helper.logError(new ProtocolTestingConnectionManagerException(
                            "the close future of channel can't be null"), signature, LOGGER);
                    }
                    openedChannels.add(channel);
                }

                // clear association maps
                channelSessionGUIDMap.clear();
                sessionGUIDChannelMap.clear();
                sessionGUIDNextCommandIndexMap.clear();
            }
            ChannelGroupFuture future = openedChannels.close();
            future.addListener(new ExceptionHandlingChannelGroupFutureListener());
            future.awaitUninterruptibly();
            channelFactory.releaseExternalResources();
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Adds association between a channel and a sessionGUID.
     * </p>
     * <p>
     * This method can be executed in multi-thread environment thus maps access has to be
     * wrapped within synchronized block.
     * </p>
     *
     * @param sessionGUID session GUID, can't be null, can't be empty
     * @param channel channel instance, can't be null.
     * @throws IllegalArgumentException - when any parameter is invalid
     */
    public void addChannelSessionGUIDAssociation(Channel channel, String sessionGUID) {
        final String signature = "BaseProtocolTestingConnectionManager#addChannelSessionGUIDAssociation";
        Helper.logEnter(signature, new String[] {"channel", "sessionGUID"},
            new Object[] {channel, sessionGUID}, LOGGER);
        try {
            Helper.checkNull(channel, "channel", signature, LOGGER);
            Helper.checkString(sessionGUID, "sessionGUID", signature, LOGGER);
            synchronized(associationMapsLock) {
                channelSessionGUIDMap.put(channel, sessionGUID);
                sessionGUIDChannelMap.put(sessionGUID, channel);
            }
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Returns sessionGUID associated with passed channel.
     * </p>
     * <p>
     * This method can be executed in multi-thread environment thus maps access has to be
     * wrapped within synchronized block.
     * </p>
     *
     * @param channel channel instance, can't be null.
     * @return Session GUID associated with given channel.
     * @throws IllegalArgumentException - when any parameter is invalid
     * @throws NoSuchSessionException - when there is no session associated with given channel
     */
    public String getSessionGUIDByChannel(Channel channel) throws NoSuchSessionException {
        final String signature = "BaseProtocolTestingConnectionManager#getSessionGUIDByChannel";
        String res = null;
        Helper.logEnter(signature, new String[] {"channel"}, new Object[] {channel}, LOGGER);
        try {
            Helper.checkNull(channel, "channel", signature, LOGGER);
            synchronized(associationMapsLock) {
                res =  channelSessionGUIDMap.get(channel);
                if (res == null) {
                    throw Helper.logError(new NoSuchSessionException(
                        "no session associated with given channel"), signature, LOGGER);
                }
            }
        } finally {
            Helper.logExit(signature, res, LOGGER);
        }
        return res;
    }

    /**
     * <p>
     * Returns channel associated with passed sessionGUID.
     * </p>
     * <p>
     * This method can be executed in multi-thread environment thus maps access
     * has to be wrapped within synchronized block.
     * </p>
     * @throws IllegalArgumentException - when any parameter is invalid
     * @throws NoSuchSessionException - when there is no session associated with given sessionGUID
     * @param sessionGUID the session GUID, can't be null/empty.
     * @return Channel associated with given sessionGUID.
     */
    public Channel getChannelBySessionGUID(String sessionGUID) throws NoSuchSessionException {
        final String signature = "BaseProtocolTestingConnectionManager#getChannelBySessionGUID";
        Helper.logEnter(signature, new String[] {"sessionGUID"}, new Object[] {sessionGUID}, LOGGER);
        Channel channel = null;
        try {
            Helper.checkString(sessionGUID, "sessionGUID", signature, LOGGER);
            synchronized(associationMapsLock) {
                channel =  sessionGUIDChannelMap.get(sessionGUID);
                if (channel == null) {
                    throw Helper.logError(new NoSuchSessionException(
                        "no session associated with given sessionGUID"), signature, LOGGER);
                }
            }
        } finally {
            Helper.logExit(signature, channel, LOGGER);
        }
        return channel;
    }

    /**
     * <p>
     * Removes association between sessionGUID and a channel.
     * </p>
     * <p>
     * This method can be executed in multi-thread environment thus maps access has
     * to be wrapped within synchronized block.
     * </p>
     * @throws IllegalArgumentException - when any parameter is invalid
     * @param channel channel instance, can't be null.
     */
    public void removeChannelSessionGUIDAssociation(Channel channel) {
        final String signature = "BaseProtocolTestingConnectionManager#removeChannelSessionGUIDAssociation";
        Helper.logEnter(signature, new String[] {"channel"}, new Object[] {channel}, LOGGER);
        try {
            Helper.checkNull(channel, "channel", signature, LOGGER);
            synchronized(associationMapsLock) {
                String sessionGUID = channelSessionGUIDMap.get(channel);
                sessionGUIDChannelMap.remove(sessionGUID);
                channelSessionGUIDMap.remove(channel);
            }
        } finally {
            Helper.logExit(signature, null, LOGGER);
        }
    }

    /**
     * <p>
     * Prepares the command to be sent by replacing the command index and timestamp in the given command.
     * </p>
     * <p>
     * This method can be executed in multi-thread environment thus maps
     * access has to be wrapped within synchronized block.
     * </p>
     * @param command the command string., can't be null/empty
     * @param channel the channel with which the command is to be sent, can't be null.
     * @return The command with command index and timestamp replaced.
     * @throws IllegalArgumentException if the command is null/empty
     * @throws IllegalMessageException when there's no commandIndex/timestamp in the command
     * @throws NoSuchSessionException when there is no session found
     */
    public String prepareCommandToSend(String command, Channel channel)
        throws IllegalMessageException, NoSuchSessionException {
        final String signature = "BaseProtocolTestingConnectionManager#prepareCommandToSend";
        Helper.logEnter(signature, new String[] {"command", "channel"}, new Object[] {command, channel}, LOGGER);

        try {
            Helper.checkString(command, "command", signature, LOGGER);
            String sessionGUID = getSessionGUIDByChannel(channel);
            Helper.checkString(sessionGUID, "sessionGUID", signature, LOGGER);
            Integer nextCommandIndex = null;
            synchronized(associationMapsLock) {
                nextCommandIndex = sessionGUIDNextCommandIndexMap.get(sessionGUID);
                if (nextCommandIndex == null) {
                    nextCommandIndex = Integer.valueOf(0);
                }
                sessionGUIDNextCommandIndexMap.put(sessionGUID, Integer.valueOf(nextCommandIndex.intValue() + 1));
            }

            StringBuffer sb = new StringBuffer();
            Matcher matcher = COMMAND_INDEX_PATTERN.matcher(command);
            if (!matcher.find()) {
                throw new IllegalMessageException("No commandIndex found");
            }

            matcher.appendReplacement(sb, "\"commandIndex\":" + nextCommandIndex.intValue());
            matcher.appendTail(sb);
            command = sb.toString();

            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.YEAR, 1900);

            sb = new StringBuffer();
            matcher = TIMESTAMP_PATTERN.matcher(command);
            if (!matcher.find()) {
                throw new IllegalMessageException("No timestamp found");
            }
            matcher.appendReplacement(sb, "\"timestamp\":" + calendar.getTime().getTime());
            matcher.appendTail(sb);
            command = sb.toString();
        } catch (NoSuchSessionException e) {
            throw Helper.logError(e, signature, LOGGER);
        } finally {
            Helper.logExit(signature, command, LOGGER);
        }
        return command;
    }

    /**
     * <p>
     * Returns if the server is started.
     * </p>
     * @return if the server is started.
     */
    private boolean isStarted() {
        return channelFactory != null && serverChannel != null;
    }
}